package cm.tang.message;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.util.Objects;

/**
 * @author Tang
 * @classname RedisMqClientProxy
 * @description [  ]
 * @date 2021/6/2 11:42
 */
public class RedisMqClientProxy {

    private RedisTemplate<String, Object> redisTemplate;

    private MessageHandler messageHandler;

    public RedisMqClientProxy(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void publish(String topic,String message) {
        this.redisTemplate.convertAndSend(topic, message);
    }

    public RedisMessageListenerContainer subscribe(String topic, IMessageHandler iMessageHandler) {
        this.messageHandler = new MessageHandler(iMessageHandler);
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(Objects.requireNonNull(this.redisTemplate.getConnectionFactory()));
        redisMessageListenerContainer.addMessageListener(this.messageHandler, new PatternTopic(topic));
        return redisMessageListenerContainer;
    }

    @Slf4j
    private static class MessageHandler implements MessageListener {

        private IMessageHandler iMessageHandler;

        public MessageHandler(IMessageHandler iMessageHandler) {
            this.iMessageHandler = iMessageHandler;
        }

        @Override
        public void onMessage(Message message, byte[] pattern) {
            String messageStr = new String(message.getBody());
            String topic = new String(pattern);
            iMessageHandler.handler(messageStr, topic);
            log.info("收到的消息内容为:{}",messageStr);
        }
    }
}